// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: DServer support simple entity
// feature: support auth token
// feature: provide query interfaces
// feature: develop dispatcher
// feature: content filter topic
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: interval report metrics
// feature: evbs support report hostname processname and version
// feature: add statictic lookup
// feature: dds support get remote participant vbs version and rm some print
// feature: Change the default order and test domain sope
// ------------------------------------------------------------------

#ifndef SRC_CPP_RTPS_PARTICIPANT_RTPSPARTICIPANTIMPL_H_
#define SRC_CPP_RTPS_PARTICIPANT_RTPSPARTICIPANTIMPL_H_
#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC

#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <edds/rtps/common/Guid.h>

#include <edds/rtps/attributes/RTPSParticipantAttributes.h>

#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>

#include <history/IChangePool.h>
#include <history/IPayloadPool.h>

#include <edds/rtps/messages/MessageReceiver.h>
#include <deps/resources/ResourceEvent.h>

#include <transport/TransportFactory.hpp>
#include <transport/ReceiverResource.h>
#include <transport/SenderResource.h>

#if HAVE_SECURITY
#include <edds/rtps/Endpoint.h>
#include <edds/rtps/security/accesscontrol/ParticipantSecurityAttributes.h>
#include <rtps/security/SecurityManager.h>
#include <edds/rtps/security/common/AuthToken.h>
#endif  // if HAVE_SECURITY

#include <ertps/utils/Semaphore.h>

#if defined(_WIN32)
#include <process.h>
#else
#include <unistd.h>
#endif  // if defined(_WIN32)

#include <sys/types.h>

#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <list>
#include <mutex>

#include <rtps/messages/RTPSMessageGroup_t.hpp>
#include <rtps/messages/SendBuffersManager.hpp>

#include <deps/common/shared_mutex.hpp>

#include "../flowcontrol/FlowControllerFactory.hpp"
#include <ertps/utils/LocatorAlias.h>
#include <xmlparser/attributes/XMLTopicAttributes.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <transport/TransportFactory.hpp>
#include <edds/rtps/common/LocatorSelector.hpp>

#include <discovery/database/DiscoveryEndpointInfo.hpp>
#include <discovery/database/DiscoveryParticipantInfo.hpp>
#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>
#include <history/ReaderListener.h>
#include <history/WriterListener.h>

using vbs::common::ReaderHistory;
using vbs::common::ReaderListener;
using vbs::common::WriterHistory;
using vbs::common::WriterListener;
namespace vbs {
namespace common {
class ParticipantListener;
}
}  // namespace vbs
namespace evbs {

namespace edds {
namespace dds {
namespace builtin {

class LookupManager;

}  // namespace builtin
}  // namespace dds
}  // namespace edds

namespace ertps {

using vbsutil::xmlparser::TopicAttributes;
class MessageReceiver;
using namespace vbs::transport;
//using vbsutil::xmlparser::string_255;
namespace rtps {

using namespace evbs::ertps::rtps::ddb;
using namespace vbs::common;

class RTPSParticipant;
//class ParticipantListener;
struct CDRMessage_t;
class Endpoint;
class RTPSWriter;
class WriterAttributes;
//class WriterHistory;
//class WriterListener;
class RTPSReader;
class ReaderAttributes;
//class ReaderHistory;
//class ReaderListener;
class StatefulReader;
class PDPSimple;
class IPersistenceService;
class WLP;
class BuiltinProtocols;

/**
 * @brief Class RTPSParticipantImpl, it contains the private implementation of the RTPSParticipant
 * functions and allows the creation and removal of writers and readers. It manages the send and
 * receive threads.
 * @ingroup RTPS_MODULE
 */
class RTPSParticipantImpl {
    /*
       Receiver Control block is a struct we use to encapsulate the resources that take part in
       message reception. It contains: -A vbs::transport::ReceiverResource (as produced by the NetworkFactory
       Element) -Its associated MessageReceiver
     */
    typedef struct ReceiverControlBlock {
        std::shared_ptr<ReceiverResource> Receiver;
        MessageReceiver* mp_receiver;  // Associated Readers/Writers inside of MessageReceiver

        ReceiverControlBlock(std::shared_ptr<ReceiverResource>& rec) : Receiver(rec), mp_receiver(nullptr) {}

        ReceiverControlBlock(ReceiverControlBlock&& origen)
            : Receiver(origen.Receiver), mp_receiver(origen.mp_receiver) {
            origen.mp_receiver = nullptr;
            origen.Receiver.reset();
        }

        void disable() {
            if (Receiver != nullptr) {
                Receiver->disable();
            }
        }

     private:
        ReceiverControlBlock(const ReceiverControlBlock&) = delete;
        const ReceiverControlBlock& operator=(const ReceiverControlBlock&) = delete;

    } ReceiverControlBlock;

 public:
    /**
     * @param param
     * @param guidP
     * @param part
     * @param plisten
     */
    RTPSParticipantImpl(uint32_t domain_id, const RTPSParticipantAttributes& param, const GuidPrefix_t& guidP,
                        RTPSParticipant* part, vbs::common::ParticipantListener* plisten = nullptr);

    /**
     * @param param
     * @param guidP
     * @param persistence_guid
     * @param part
     * @param plisten
     */
    RTPSParticipantImpl(uint32_t domain_id, const RTPSParticipantAttributes& param, const GuidPrefix_t& guidP,
                        const GuidPrefix_t& persistence_guid, RTPSParticipant* part,
                        vbs::common::ParticipantListener* plisten = nullptr);

    virtual ~RTPSParticipantImpl();

    // Create receiver resources and start builtin protocols
    void enable();

    // Stop builtin protocols and delete receiver resources
    void disable();

    /**
     * Get associated GUID
     * @return Associated GUID
     */
    inline const GUID_t& getGuid() const { return m_guid; }

    void setGuid(GUID_t& guid);

    //! Announce RTPSParticipantState (force the sending of a DPD message.)
    void announceRTPSParticipantState();

    //! Stop the RTPSParticipant Announcement (used in tests to avoid multiple packets being send)
    void stopRTPSParticipantAnnouncement();

    //! Reset to timer to make periodic RTPSParticipant Announcements.
    void resetRTPSParticipantAnnouncement();

    // Start to send announcement message with initial announcement period.
    bool startRTPSParticipantInitialAnnouncement(uint32_t count, ertps::Duration_t period);

    void loose_next_change();

    /**
     * Activate a Remote Endpoint defined in the Static Discovery.
     * @param pguid GUID_t of the endpoint.
     * @param userDefinedId userDeinfed Id of the endpoint.
     * @param kind kind of endpoint
     * @return True if correct.
     */
    bool newRemoteEndpointDiscovered(const GUID_t& pguid, int16_t userDefinedId, EndpointKind_t kind);

    /**
     * Assert the liveliness of a remote participant
     * @param remote_guid GuidPrefix_t of the participant.
     */
    void assert_remote_participant_liveliness(const GuidPrefix_t& remote_guid);

    /**
     * Get the RTPSParticipant ID
     * @return RTPSParticipant ID
     */
    inline uint32_t getRTPSParticipantID() const { return (uint32_t)m_att.participantID; }

    //! Post to the resource semaphore
    void ResourceSemaphorePost();

    //! Wait for the resource semaphore
    void ResourceSemaphoreWait();

    //! Get Pointer to the Event Resource.
    vbs::common::ResourceEvent& getEventResource() { return mp_event_thr; }

    /**
     * Send a message to several locations
     * @param msg Message to send.
     * @param sender_guid GUID of the producer of the message.
     * @param destination_locators_begin Iterator at the first destination locator.
     * @param destination_locators_end Iterator at the end destination locator.
     * @param max_blocking_time_point execution time limit timepoint.
     * @return true if at least one locator has been sent.
     */
    template <class LocatorIteratorT>
    bool sendSync(CDRMessage_t* msg, const GUID_t& /*sender_guid*/, const LocatorIteratorT& destination_locators_begin,
                  const LocatorIteratorT& destination_locators_end, bool /*is_send_multi = false*/) {
        bool ret_code = false;

        if (m_transportFactory) {
            ret_code = m_transportFactory->Send(get_transport_name(), msg->buffer, msg->length,
                                                destination_locators_begin, destination_locators_end);
        }

        return ret_code;
    }

    //! Get the participant Mutex
    std::recursive_mutex* getParticipantMutex() const { return mp_mutex; }

    /**
     * Get the participant listener
     * @return participant listener
     */
    inline vbs::common::ParticipantListener* getListener() {
        std::lock_guard<std::recursive_mutex> _(*getParticipantMutex());
        return mp_participantListener;
    }

    /**
     * @brief Modifies the participant listener
     * @param listener
     */
    void set_listener(vbs::common::ParticipantListener* listener) {
        std::lock_guard<std::recursive_mutex> _(*getParticipantMutex());
        mp_participantListener = listener;
    }

    std::vector<std::string> getParticipantNames() const;

    /**
     * Get the participant
     * @return participant
     */
    inline RTPSParticipant* getUserRTPSParticipant() { return mp_userParticipant; }

    /*!
     * @remarks Non thread-safe.
     */
    const std::vector<RTPSWriter*>& getAllWriters() const;

    /*!
     * @remarks Non thread-safe.
     */
    const std::vector<RTPSReader*>& getAllReaders() const;

    uint32_t getMaxMessageSize() const;

    uint32_t getMaxDataSize();

    uint32_t calculateMaxDataSize(uint32_t length);

#if HAVE_SECURITY
    security::SecurityManager& security_manager() { return m_security_manager; }

    const security::ParticipantSecurityAttributes& security_attributes() { return security_attributes_; }

    inline bool is_security_initialized() const { return m_security_manager.is_security_initialized(); }

    inline bool is_secure() const { return m_security_manager.is_security_active(); }

    bool pairing_remote_reader_with_local_writer_after_security(const GUID_t& local_writer,
                                                                const ReaderProxyData& remote_reader_data);

    bool pairing_remote_writer_with_local_reader_after_security(const GUID_t& local_reader,
                                                                const WriterProxyData& remote_writer_data);

    /**
     * @brief Checks whether the writer has security attributes enabled
     * @param writer_attributes Attibutes of the writer as given to the create_writer
     */
    bool is_security_enabled_for_writer(const WriterAttributes& writer_attributes);

    /**
     * @brief Checks whether the reader has security attributes enabled
     * @param reader_attributes Attibutes of the reader as given to the create_reader
     */
    bool is_security_enabled_for_reader(const ReaderAttributes& reader_attributes);

    security::AuthToken& auth_token() { return m_auth_token; }

#endif  // if HAVE_SECURITY
    PDP* pdp();

    PDPSimple* pdpsimple();

    WLP* wlp();

    edds::dds::builtin::LookupManager* lookup_manager() const;

    bool is_intraprocess_only() const { return is_intraprocess_only_; }

    vbs::transport::TransportFactory* network_factory() { return m_transportFactory.get(); }

    const vbs::transport::TransportFactory* network_factory() const { return m_transportFactory.get(); }

    inline bool has_shm_transport() { return has_shm_transport_; }

    uint32_t get_min_network_send_buffer_size() { return m_minSendBufferSize; }

    /**
     * Get the list of locators from which this participant may send data.
     *
     * @param [out] locators  LocatorList_t where the list of locators will be stored.
     */
    void get_sending_locators(rtps::LocatorList_t& locators) const;

    const Locator_t& find_optimal_remote_locator(const ResourceLimitedVector<Locator_t>& locators) const;
    bool is_valid_remote_locator(rtps::Locator_t& locator) const;
    /***
     * @returns A pointer to a local reader given its endpoint guid, or nullptr if not found.
     */
    RTPSReader* find_local_reader(const GUID_t& reader_guid);

    /***
     * @returns A pointer to a local writer given its endpoint guid, or nullptr if not found.
     */
    RTPSWriter* find_local_writer(const GUID_t& writer_guid);

    /**
     * @brief Fills a new entityId if set to unknown, or checks if a entity already exists with that
     * entityId in other case.
     * @param entityId to check of fill. If filled, EntityKind will be "vendor-specific" (0x01)
     * @return True if filled or the entityId is available.
     */
    bool get_new_entity_id(EntityId_t& entityId);

    void set_check_type_function(std::function<bool(const std::string&)>&& check_type);

    bool check_type(const std::string& type_name) {
        if (type_check_fn_ != nullptr) {
            return type_check_fn_(type_name);
        }
        return false;
    }

    RTPSMessageGroup_t* get_send_buffer() { return send_buffers_->get_buffer(this); }

    void return_send_buffer(RTPSMessageGroup_t* buffer) { send_buffers_->return_buffer(buffer); }

    uint32_t get_domain_id() const { return domain_id_; }

    //! Compare metatraffic locators list searching for mutations
    bool did_mutation_took_place_on_meta(const LocatorList_t& MulticastLocatorList,
                                         const LocatorList_t& UnicastLocatorList) const;

    // //! Getter client_override flag
    // bool client_override() {
    //     return client_override_;
    // }
    // //! Setter client_override flag
    // void client_override(bool value) {
    //     client_override_ = value;
    // }

    //! Retrieve persistence guid prefix
    GuidPrefix_t get_persistence_guid_prefix() const { return m_persistence_guid.guidPrefix; }

    bool is_initialized() const { return initialized_; }

    //! Set remote evbs version
    void set_remote_evbsversion(EvbsVersion_t evbsversion) { m_evbsRemoteVersion = evbsversion; }

    //! Get remote evbs version
    EvbsVersion_t get_remote_evbsversion();

    //! Set remote evbs version
    void set_remote_hostname(const string_255& hostname) { m_hostName = hostname; }

    EvbsVersion_t get_version_by_guid(const GuidPrefix_t& guid_prefix);

    /**
     * @brief Query if the participant is found in the ignored collection
     *
     * @param [in] participant_guid Participant to be queried
     * @return True if found in the ignored collection. False otherwise.
     */
    bool is_participant_ignored(const GuidPrefix_t& participant_guid);

    /**
     * @brief Query if the writer is found in the ignored collection
     *
     * @param [in] writer_guid Writer to be queried
     * @return True if found in the ignored collection. False otherwise.
     */
    bool is_writer_ignored(const GUID_t& writer_guid);

    /**
     * @brief Query if the reader is found in the ignored collection
     *
     * @param [in] reader_guid Reader to be queried
     * @return True if found in the ignored collection. False otherwise.
     */
    bool is_reader_ignored(const GUID_t& reader_guid);

    /**
     * @brief Add a Participant into the corresponding ignore collection.
     *
     * @param [in] participant_guid Participant that is to be ignored.
     * @return True if correctly included into the ignore collection. False otherwise.
     */
    bool ignore_participant(const GuidPrefix_t& participant_guid);

    /**
     * @brief Add a Writer into the corresponding ignore collection.
     *
     * @param [in] writer_guid Writer that is to be ignored.
     * @return True if correctly included into the ignore collection. False otherwise.
     */
    bool ignore_writer(const GUID_t& writer_guid);

    /**
     * @brief Add a Reader into the corresponding ignore collection.
     *
     * @param [in] reader_guid Reader that is to be ignored.
     * @return True if correctly included into the ignore collection. False otherwise.
     */
    bool ignore_reader(const GUID_t& reader_guid);
    void select_locators(LocatorSelector& selector);
    bool RegisterTransport(TransmitorType type);
    bool bTransportRegistered(TransmitorType type);

    std::string get_db_info(bool log_record);

    // Get all the participants in DataBase
    bool get_participants_in_db(std::vector<GuidPrefix_t>& prefix_list);

    // Get the participant info by guid_prefix
    bool get_participant_info(const GuidPrefix_t& prefix, DiscoveryParticipantInfo& part_info);

    // Get the reader info by guid
    bool get_reader_info(const GUID_t& reader_guid, DiscoveryEndpointInfo& reader_info);

    // Get the writer info by guid
    bool get_writer_info(const GUID_t& writer_guid, DiscoveryEndpointInfo& writer_info);

    bool get_topic_names(std::vector<std::string>& topic_list);

    bool has_writer(const std::string& topic_name);

    bool has_reader(const std::string& topic_name);

    bool get_writer_of_topic(const std::string& topic_name, std::vector<DiscoveryEndpointInfo>& writer_info_list);

    bool get_reader_of_topic(const std::string& topic_name, std::vector<DiscoveryEndpointInfo>& reader_info_list);

    bool has_participant(const std::string& participant_name);

    bool get_participant_of_name(const std::string& participant_name, DiscoveryParticipantInfo& part_info);

    void registed_user_transport(vbs::RemoteEndpointInfo::LocatorType type, bool registed);

    bool registed_user_transport(vbs::RemoteEndpointInfo::LocatorType type);

 private:
    //! Indicates whether the participant has shared-memory transport
    bool has_shm_transport_;

    //! DomainId
    uint32_t domain_id_;
    //! Attributes of the RTPSParticipant
    RTPSParticipantAttributes m_att;
    //! Guid of the RTPSParticipant.
    GUID_t m_guid;
    // //! String containing the RTPSParticipant Guid.
    // std::string guid_str_;
    //! Persistence guid of the RTPSParticipant
    GUID_t m_persistence_guid;
    //! Sending resources. - DEPRECATED -Stays commented for reference purposes
    // ResourceSend* mp_send_thr;
    //! Event Resource
    vbs::common::ResourceEvent mp_event_thr;
    //! BuiltinProtocols of this RTPSParticipant
    BuiltinProtocols* mp_builtinProtocols;
    // //! Semaphore to wait for the listen thread creation.
    // Semaphore* mp_ResourceSemaphore;
    //! Id counter to correctly assign the ids to writers and readers.
    std::atomic<uint32_t> IdCounter;

    std::shared_ptr<vbs::transport::TransportFactory> m_transportFactory;
    bool b_registed_transport =
        false;  //用来判断是否要在析构时解注册transport，相同guid的participant创建2次，创建失败的participant解注册transport会导致并发异常
    uint32_t m_maxMessageSize = std::numeric_limits<uint32_t>::max();
    uint32_t m_minSendBufferSize = std::numeric_limits<uint32_t>::max();

    /* 判断是否注册过UDP Transport */
    std::atomic<bool> registed_user_transport_[2] {false};

    //! Maximum number of bytes allowed for an RTPS datagram generated by this writer.
    uint32_t max_output_message_size_ = 1400;  //std::numeric_limits<uint32_t>::max();

    //! Mutex to safely access endpoints collections
    mutable vbs::shared_mutex endpoints_list_mutex;
    //! Writer List.
    std::vector<RTPSWriter*> m_allWriterList;
    //! Reader List
    std::vector<RTPSReader*> m_allReaderList;
    //! Listen thread list.
    //! Writer List.
    std::vector<RTPSWriter*> m_userWriterList;
    //! Reader List
    std::vector<RTPSReader*> m_userReaderList;

    //! Type cheking function
    std::function<bool(const std::string&)> type_check_fn_;
    //! Pool of send buffers
    std::unique_ptr<SendBuffersManager> send_buffers_;

    // /**
    //  * Client override flag: SIMPLE participant that has been overriden with the environment
    //  * variable and transformed into a client.
    //  */
    // bool client_override_;

    //! Autogenerated metatraffic locators flag
    bool internal_metatraffic_locators_;
    //! Autogenerated default locators flag
    bool internal_default_locators_;

#if HAVE_SECURITY
    // Security manager
    security::SecurityManager m_security_manager;
    security::AuthToken m_auth_token;
#endif  // if HAVE_SECURITY

    //! Encapsulates all associated resources on a Receiving element.
    std::list<ReceiverControlBlock> m_receiverResourcelist;
    //! Receiver resource list needs its own mutext to avoid a race condition.
    vbs::shared_mutex m_receiverResourcelistMutex;

    //! Participant Listener
    vbs::common::ParticipantListener* mp_participantListener;
    //! Pointer to the user participant
    RTPSParticipant* mp_userParticipant;

    //! Determine if the RTPSParticipantImpl was initialized successfully.
    bool initialized_ = false;

    //! evbs Version
    EvbsVersion_t m_evbsRemoteVersion;

    //! host name
    string_255 m_hostName;

    //! Ignored entities collections
    std::set<GuidPrefix_t> ignored_participants_;
    std::set<GUID_t> ignored_writers_;
    std::set<GUID_t> ignored_readers_;
    //! Protect ignored entities collection concurrent access
    mutable vbs::shared_mutex ignored_mtx_;

    RTPSParticipantImpl& operator=(const RTPSParticipantImpl&) = delete;

    /**
     * Method to check if a specific entityId already exists in this RTPSParticipant
     * @param ent EnityId to check
     * @param kind Endpoint Kind.
     * @return True if exists.
     */
    bool existsEntityId(const EntityId_t& ent, EndpointKind_t kind) const;

    /**
     * Assign an endpoint to the vbs::transport::ReceiverResources, based on its LocatorLists.
     * @param endp Pointer to the endpoint.
     * @return True if correct.
     */
    bool assignEndpointListenResources(Endpoint* endp);

    /** Assign an endpoint to the vbs::transport::ReceiverResources as specified specifically on parameter list
     * @param pend Pointer to the endpoint.
     * @param lit Locator list iterator.
     * @param isMulticast Boolean indicating that is multicast.
     * @param isFixed Boolean indicating that is a fixed listenresource.
     * @return True if assigned.
     */
    bool assignEndpoint2LocatorList(Endpoint* pend, LocatorList_t& list);

    /** Create the new vbs::transport::ReceiverResources needed for a new Locator, contains the calls to
       assignEndpointListenResources and consequently assignEndpoint2LocatorList
        @param pend - Pointer to the endpoint which triggered the creation of the Receivers.
        @param unique_flows - Whether unique listening ports should be created for this endpoint.
        @param initial_unique_port - First unique listening port to try.
        @param final_unique_port - Unique listening port that will not be tried.
     */
    bool createAndAssociateReceiverswithEndpoint(Endpoint* pend, bool unique_flows = false,
                                                 uint16_t initial_unique_port = 0U, uint16_t final_unique_port = 0U);

    /** Create non-existent SendResources based on the Locator list of the entity
        @param pend - Pointer to the endpoint whose SenderResources are to be created
     */
    bool createSendResources(Endpoint* pend);

    /** When we want to create a new Resource but the physical channel specified by the Locator
        can not be opened, we want to mutate the Locator to open a more or less equivalent channel.
        @param loc -  Locator we want to change
     */
    Locator_t& applyLocatorAdaptRule(Locator_t& loc);

    /**
     * Update port for all endpoint locators when it has a value of 0 and then
     * apply locator normalization.
     *
     * @param [in, out] endpoint_att  EndpointAttributes to be updated
     */
    void normalize_endpoint_locators(EndpointAttributes& endpoint_att);

    //! Participant Mutex
    std::recursive_mutex* mp_mutex;

    //! Will this participant use intraprocess only?
    bool is_intraprocess_only_;

    /*
     * Flow controller factory.
     */
    edds::rtps::FlowControllerFactory flow_controller_factory_;

#if HAVE_SECURITY
    security::ParticipantSecurityAttributes security_attributes_;
#endif  // if HAVE_SECURITY

    /**
     * Get persistence service from factory, using endpoint attributes (or participant
     * attributes if endpoint does not define a persistence service config)
     */
    IPersistenceService* get_persistence_service(const EndpointAttributes& param);

    /**
     * Returns the Durability kind from which a endpoint is able to use the persistence service.
     */
    DurabilityKind_t get_persistence_durability_red_line(bool is_builtin_endpoint);

    /**
     * Check if persistence is required and return persistence service from factory,
     * using endpoint attributes (or participant
     * attributes if endpoint does not define a persistence service config)
     *
     * @param [in]  is_builtin  Whether the enpoint being created is a builtin one.
     * @param [in]  param       Attributes of the endpoint being created.
     * @param [out] service     Pointer to the persistence service.
     *
     * @return false if parameters are not consistent or the service should be created and couldn't
     * @return true if persistence service is not required
     * @return true if persistence service is created
     */
    bool get_persistence_service(bool is_builtin, const EndpointAttributes& param, IPersistenceService*& service);

    template <typename Functor>
    bool create_writer(RTPSWriter** writer_out, WriterAttributes& param, const EntityId_t& entity_id, bool is_builtin,
                       const Functor& callback);

    template <typename Functor>
    bool create_reader(RTPSReader** reader_out, ReaderAttributes& param, const EntityId_t& entity_id, bool is_builtin,
                       bool enable, const Functor& callback);

    /**
     * Get default metatraffic locators when not provided by the user.
     */
    void get_default_metatraffic_locators();

    /**
     * Get default unicast locators when not provided by the user.
     */
    void get_default_unicast_locators();

 public:
    const RTPSParticipantAttributes& getRTPSParticipantAttributes() const { return this->m_att; }

    /**
     * Create a Writer in this RTPSParticipant.
     * @param Writer Pointer to pointer of the Writer, used as output. Only valid if return==true.
     * @param param WriterAttributes to define the Writer.
     * @param hist Pointer to the WriterHistory.
     * @param listen Pointer to the WriterListener.
     * @param entityId EntityId assigned to the Writer.
     * @param isBuiltin Bool value indicating if the Writer is builtin (Discovery or Liveliness
     * protocol) or is created for the end user.
     * @return True if the Writer was correctly created.
     */
    bool createWriter(RTPSWriter** Writer, WriterAttributes& param, WriterHistory* hist, WriterListener* listen,
                      const EntityId_t& entityId = c_EntityId_Unknown, bool isBuiltin = false);

    /**
     * Create a Writer in this RTPSParticipant with a custom payload pool.
     * @param Writer Pointer to pointer of the Writer, used as output. Only valid if return==true.
     * @param param WriterAttributes to define the Writer.
     * @param payload_pool Shared pointer to the IPayloadPool
     * @param hist Pointer to the WriterHistory.
     * @param listen Pointer to the WriterListener.
     * @param entityId EntityId assigned to the Writer.
     * @param isBuiltin Bool value indicating if the Writer is builtin (Discovery or Liveliness
     * protocol) or is created for the end user.
     * @return True if the Writer was correctly created.
     */
    bool createWriter(RTPSWriter** Writer, WriterAttributes& param, const std::shared_ptr<IPayloadPool>& payload_pool,
                      WriterHistory* hist, WriterListener* listen, const EntityId_t& entityId = c_EntityId_Unknown,
                      bool isBuiltin = false);

    /**
     * Create a Writer in this RTPSParticipant with a custom payload pool.
     * @param Writer Pointer to pointer of the Writer, used as output. Only valid if return==true.
     * @param watt WriterAttributes to define the Writer.
     * @param payload_pool Shared pointer to the IPayloadPool
     * @param change_pool Shared pointer to the IChangePool
     * @param hist Pointer to the WriterHistory.
     * @param listen Pointer to the WriterListener.
     * @param entityId EntityId assigned to the Writer.
     * @param isBuiltin Bool value indicating if the Writer is builtin (Discovery or Liveliness
     * protocol) or is created for the end user.
     * @return True if the Writer was correctly created.
     */
    bool createWriter(RTPSWriter** Writer, WriterAttributes& watt, const std::shared_ptr<IPayloadPool>& payload_pool,
                      const std::shared_ptr<IChangePool>& change_pool, WriterHistory* hist, WriterListener* listen,
                      const EntityId_t& entityId = c_EntityId_Unknown, bool isBuiltin = false);

    /**
     * Create a Reader in this RTPSParticipant.
     * @param Reader Pointer to pointer of the Reader, used as output. Only valid if return==true.
     * @param param ReaderAttributes to define the Reader.
     * @param hist Pointer to the ReaderHistory.
     * @param listen Pointer to the ReaderListener.
     * @param entityId EntityId assigned to the Reader.
     * @param isBuiltin Bool value indicating if the Reader is builtin (Discovery or Liveliness
     * protocol) or is created for the end user.
     * @param enable Whether the reader should be automatically enabled.
     * @return True if the Reader was correctly created.
     */
    bool createReader(RTPSReader** Reader, ReaderAttributes& param, ReaderHistory* hist, ReaderListener* listen,
                      const EntityId_t& entityId = c_EntityId_Unknown, bool isBuiltin = false, bool enable = true);

    /**
     * Create a Reader in this RTPSParticipant with a custom payload pool.
     * @param Reader Pointer to pointer of the Reader, used as output. Only valid if return==true.
     * @param param ReaderAttributes to define the Reader.
     * @param payload_pool Shared pointer to the IPayloadPool
     * @param hist Pointer to the ReaderHistory.
     * @param listen Pointer to the ReaderListener.
     * @param entityId EntityId assigned to the Reader.
     * @param isBuiltin Bool value indicating if the Reader is builtin (Discovery or Liveliness
     * protocol) or is created for the end user.
     * @param enable Whether the reader should be automatically enabled.
     * @return True if the Reader was correctly created.
     */
    bool createReader(RTPSReader** Reader, ReaderAttributes& param, const std::shared_ptr<IPayloadPool>& payload_pool,
                      ReaderHistory* hist, ReaderListener* listen, const EntityId_t& entityId = c_EntityId_Unknown,
                      bool isBuiltin = false, bool enable = true);

    bool createReader(RTPSReader** Reader, ReaderAttributes& param, const std::shared_ptr<IChangePool>& change_pool,
                      ReaderHistory* hist, ReaderListener* listen, const EntityId_t& entityId = c_EntityId_Unknown,
                      bool isBuiltin = false, bool enable = true);

    bool enableReader(RTPSReader* reader);

    void disableReader(RTPSReader* reader);

    /**
     * Register a Writer in the BuiltinProtocols.
     * @param Writer Pointer to the RTPSWriter.
     * @param topicAtt TopicAttributes of the Writer.
     * @param wqos WriterQos.
     * @return True if correctly registered.
     */
    bool registerWriter(RTPSWriter* Writer, const TopicAttributes& topicAtt, const WriterQos& wqos,
                        const vbsutil::xmlparser::ContentFilterProperty* content_filter = nullptr);

    /**
     * Register a Reader in the BuiltinProtocols.
     * @param Reader          Pointer to the RTPSReader.
     * @param topicAtt        TopicAttributes of the Reader.
     * @param rqos            ReaderQos.
     * @return True if correctly registered.
     */
    bool registerReader(RTPSReader* Reader, const TopicAttributes& topicAtt, const ReaderQos& rqos,
                        const vbsutil::xmlparser::ContentFilterProperty* content_filter = nullptr);

    /**
     * Update participant attributes.
     * @param patt New participant attributes.
     * @return True on success, false otherwise.
     */
    void update_attributes(const RTPSParticipantAttributes& patt);

    /**
     * Update local writer QoS
     * @param Writer Writer to update
     * @param wqos New QoS for the writer
     * @return True on success
     */
    bool updateLocalWriter(RTPSWriter* Writer, const TopicAttributes& topicAtt, const WriterQos& wqos,
                           const vbsutil::xmlparser::ContentFilterProperty* content_filter = nullptr,
                           bool notify = true);

    /**
     * Update local reader QoS
     * @param Reader          Reader to update
     * @param topicAtt        TopicAttributes of the Reader.
     * @param rqos            New QoS for the reader
     * @return True on success
     */
    bool updateLocalReader(RTPSReader* Reader, const TopicAttributes& topicAtt, const ReaderQos& rqos,
                           const vbsutil::xmlparser::ContentFilterProperty* content_filter = nullptr,
                           bool notify = true);

    /**
     * Get the participant attributes
     * @return Participant attributes
     */
    inline RTPSParticipantAttributes& getAttributes() { return m_att; }

    /**
     * Get the domain id from port
     * @return domain id
     */
    inline DomainId_t getDomainIdFromePort(uint16_t port) { return m_att.port.getDomainId(port); }
    /**
     * Delete a user endpoint
     * @param Endpoint to delete
     * @return True on success
     */
    bool deleteUserEndpoint(const GUID_t&);

    //! Delete all user endpoints, builtin are disposed in its related classes
    void deleteAllUserEndpoints();

    /** Traverses the user writers collection transforming its elements with a provided functor
     * @param f - Functor applied to each element. Must accept a reference as parameter. Should
     * return true to keep iterating.
     * @return Functor provided in order to allow aggregates retrieval
     */
    template <class Functor>
    Functor forEachUserWriter(Functor f) {
        // check if we are reentrying
        vbs::shared_lock<vbs::shared_mutex> _(endpoints_list_mutex);

        // traverse the list
        for (RTPSWriter* pw : m_userWriterList) {
            if (!f(*pw)) {
                break;
            }
        }

        return f;
    }

    /** Traverses the user readers collection transforming its elements with a provided functor
     * @param f - Functor applied to each element. Must accept a reference as parameter. Should
     * return true to keep iterating.
     * @return Functor provided in order to allow aggregates retrieval
     */
    template <class Functor>
    Functor forEachUserReader(Functor f) {
        // check if we are reentrying
        vbs::shared_lock<vbs::shared_mutex> _(endpoints_list_mutex);

        for (RTPSReader* pr : m_userReaderList) {
            if (!f(*pr)) {
                break;
            }
        }

        return f;
    }

    /** Helper function that creates vbs::transport::ReceiverResources based on a Locator_t List, possibly mutating
     * some and updating the list. DOES NOT associate endpoints with it.
     * @param Locator_list - Locator list to be used to create the vbs::transport::ReceiverResources
     * @param ApplyMutation - True if we want to create a Resource with a "similar" locator if the
     * one we provide is unavailable
     * @param RegisterReceiver - True if we want the receiver to be registered. Useful for receivers
     * created after participant is enabled.
     */
    bool createReceiverResources(LocatorList_t& Locator_list, bool ApplyMutation, bool RegisterReceiver,
                                 int32_t kind = LOCATOR_KIND_INVALID);

    void createSenderResources(const LocatorList_t& locator_list);

    void createSenderResources(const Locator_t& locator);

    /**
     * Function run when the RTPSDomain is notified that the environment file has changed.
     */
    void environment_file_has_changed();

    template <EndpointKind_t kind, octet no_key, octet with_key>
    static bool preprocess_endpoint_attributes(const EntityId_t& entity_id, std::atomic<uint32_t>& id_count,
                                               EndpointAttributes& att, EntityId_t& entId);

#if HAVE_SECURITY
    void set_endpoint_rtps_protection_supports(Endpoint* endpoint, bool support) {
        endpoint->supports_rtps_protection_ = support;
    }

#endif  // if HAVE_SECURITY

    std::string get_transport_name() const;  //name register in transport, unique for each participant in same process.

    ReturnCode_t assert_liveliness(RTPSWriter* writer);
    ReturnCode_t assert_liveliness();
};

}  // namespace rtps
}  // namespace ertps
} /* namespace evbs */
#endif  // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC
#endif  // SRC_CPP_RTPS_PARTICIPANT_RTPSPARTICIPANTIMPL_H_
